/*
 * Copyright (c) 2005 - 2010 CAS Dev Team
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 4. Neither the name of the CAS Dev. Team nor the names of its contributors
 *    may be used to endorse or promote products derived from this software
 *    without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 *      ASMemcachedObject.cpp
 *
 * $CAS$
 */

#include "ASMemcachedObject.hpp"
#include <ASException.hpp>

#include <CTPP2Util.hpp>
#include <CTPP2JSONParser.hpp>

#include <sys/time.h>
#include <stdio.h>
#include <zlib.h>

#define DEFAULT_COMPRESSION_LEVEL 9
#define DEFAULT_MEM_LEVEL         9
#define DEFAULT_BUFFER_SIZE       (64 * 1024)

namespace CAS_MOD
{

//
// Compress data
//
static INT_32 CompressData(const STLW::string & sSource,
                           STLW::string       & sDestination)
{
	if (sSource.empty()) { return 0; }

	int iCompressionLevel = DEFAULT_COMPRESSION_LEVEL;
	int iMemLevel         = DEFAULT_MEM_LEVEL;

	z_stream oZStream;
	memset(&oZStream, 0, sizeof(z_stream));

	int zRC = deflateInit2(&oZStream, iCompressionLevel, Z_DEFLATED, (-MAX_WBITS), iMemLevel, Z_DEFAULT_STRATEGY);
	if (zRC != Z_OK)
	{
		fprintf(stderr, "Can't init zlib: %s\n", oZStream.msg);
		deflateEnd(&oZStream);
		return -1;
	}

	oZStream.avail_in  = sSource.size();
	oZStream.avail_out = oZStream.avail_in + oZStream.avail_in / 1000 + 12 + 1;

	Byte * pInput  = (Byte *)sSource.data();
	Byte * pOutput = (Byte *)malloc(oZStream.avail_out);

	oZStream.next_in  = pInput;
	oZStream.next_out = pOutput;

	if (pOutput == NULL)
	{
		fprintf(stderr, "Can't allocate memory for buffer");
		deflateEnd(&oZStream);
		return -1;
	}

	zRC = deflate(&oZStream, Z_FINISH);
	if (zRC != Z_STREAM_END)
	{
		fprintf(stderr, "Can't compress data: %s\n", oZStream.msg);
		free(pOutput);
		deflateEnd(&oZStream);
	}

//fprintf(stderr, "CompressData %d -> %d\n", INT_32(sSource.size()), INT_32(oZStream.total_out));

	sDestination.assign((CHAR_P)pOutput, oZStream.total_out);
	free(pOutput);

	deflateEnd(&oZStream);

return sDestination.size();
}

//
// Decompress data
//
static INT_32 DecompressData(const STLW::string & sSource,
                             STLW::string       & sDestination)
{
	if (sSource.empty()) { return 0; }

	z_stream oZStream;
	memset(&oZStream, 0, sizeof(z_stream));

	int zRC = inflateInit2(&oZStream, (-MAX_WBITS));
	if (zRC != Z_OK)
	{
		fprintf(stderr, "Can't init zlib: %s\n", oZStream.msg);
		inflateEnd(&oZStream);
		return -1;
	}

	char vBuffer[DEFAULT_BUFFER_SIZE];

	INT_32 iAvailableBytes = sSource.size();
	INT_32 iPos = 0;

	oZStream.avail_in = sSource.size();
	oZStream.next_in = (Byte *)sSource.data();

        // Ok, get data
        oZStream.next_in  = (Bytef *)sSource.data();
        oZStream.avail_in = sSource.size();

        do
        {
		oZStream.next_out  = (Bytef*)vBuffer;
		oZStream.avail_out = DEFAULT_BUFFER_SIZE;
//fprintf(stderr, "D avail_in %d avail_out %d", oZStream.avail_in, oZStream.avail_out);
		zRC = inflate(&oZStream, Z_NO_FLUSH);
		if (zRC == Z_NEED_DICT || zRC == Z_DATA_ERROR || zRC == Z_MEM_ERROR)
		{
			fprintf(stderr, "Can't inflate zlib: %s\n", oZStream.msg);
			inflateEnd(&oZStream);
			return -1;
		}
//fprintf(stderr, " -> avail_in %d avail_out %d\n", oZStream.avail_in, oZStream.avail_out);
		INT_32 iLen = DEFAULT_BUFFER_SIZE - oZStream.avail_out;
		if (iLen != 0) { sDestination.append((char*)vBuffer, iLen); }
	}
	while (oZStream.avail_out == 0);

	inflateEnd(&oZStream);

//	fprintf(stderr, "DecompressData %d -> %d\n", INT_32(sSource.size()), INT_32(sDestination.size()));
return sDestination.size();
}

//
// Compress value
//
void ASMemcachedObject::CompressValue(STLW::string  & sSerializedValue,
                                      UINT_64       & iFlags) const
{
	// Check compression threshold
	if(iCompressThreshold > 0 && sSerializedValue.size() > iCompressThreshold)
	{
		STLW::string sTMP;
		CompressData(sSerializedValue, sTMP);
		// Check compression ratio
		if (dCompressRatio != 0 && 1.0 * sTMP.size() / sSerializedValue.size() < dCompressRatio)
		{
			iFlags |= 0x10;
			sSerializedValue = sTMP;
		}
	}
}

//
// Decompress value
//
void ASMemcachedObject::DecompressValue(STLW::string  & sSerializedValue,
                                        UINT_64       & iFlags) const
{
	// Check compression rate
	if(iFlags & 0x10)
	{
		STLW::string sTMP;
		DecompressData(sSerializedValue, sTMP);

		sSerializedValue = sTMP;
	}
}

//
// Serialize CDT to string with specified serialization method
//
INT_64 ASMemcachedObject::SerializeCDT(const CTPP::CDT  & oCDT,
                                       STLW::string     & sData) const
{
	if (oCDT.GetType() <= CTPP::CDT::STRING_VAL) { sData = oCDT.GetString(); return 0; }

	UINT_64 iFlags = 0;
	switch(eSerializationMethod)
	{
		case NONE:
			sData = oCDT.GetString();
			iFlags = 0;
			break;

		case JSON:
			CTPP::CDT2JSON(oCDT, sData);
			iFlags = 1;
			break;

		case PERL_STORABLE:
		case PHP_SERIALIZE:
		case PYTHON_MARSHALL:
		default:
			throw LogicError("Invalid serialization method");
	}

	CompressValue(sData, iFlags);

return iFlags;
}

//
// Un-Serialize CDT to string with specified serialization method
//
void ASMemcachedObject::RestoreCDT(const STLW::string  & sData,
                                   CTPP::CDT           & oCDT) const
{

	switch(eSerializationMethod)
	{
		case NONE:
			oCDT = sData;
			return;

		case JSON:
			{
				CTPP::CTPP2JSONParser oJSONParser(oCDT);
				oJSONParser.Parse(sData.data(), sData.data() + sData.size());
			}
			return;

		case PERL_STORABLE:
		case PHP_SERIALIZE:
		case PYTHON_MARSHALL:
		default:
			throw LogicError("Invalid serialization method");
	}
}

//
// Constructor
//
ASMemcachedObject::ASMemcachedObject(const eSerializer  & eISerializationMethod,
                                     const INT_32         iIErrorTimeout,
                                     const UINT_32        iMaxContinuumElements,
                                     const INT_32         iICompressThreshold,
                                     const W_FLOAT        dICompressRatio): eSerializationMethod(eISerializationMethod),
                                                                            iErrorTimeout(iIErrorTimeout),
                                                                            oMemcachedList(iMaxContinuumElements),
                                                                            iCompressThreshold(iICompressThreshold),
                                                                            dCompressRatio(dICompressRatio)
{
	;;
}

//
// Add memcached server
//
INT_32 ASMemcachedObject::AddServer(const STLW::string  & sHost,
                                    const INT_32          iPort,
                                    const INT_32          iProto,
                                    const W_FLOAT         dWeight,
                                    const INT_32          iConnTimeout,
                                    const INT_32          iIOTimeout)
{
	MemcachedData oMemcachedData;
	oMemcachedData.host          = sHost;
	oMemcachedData.port          = iPort;
	oMemcachedData.proto         = iProto;
	oMemcachedData.weight        = dWeight;
	oMemcachedData.last_error_ts = 0;
	oMemcachedData.hash_id       = -1;

	oMemcachedData.memcached = new Memcached(sHost, iPort, iProto, iConnTimeout, iIOTimeout);

	vServers.push_back(oMemcachedData.memcached);

	CHAR_8 szTMP[256];
	snprintf(szTMP, 256, "%s:%u_%f", sHost.c_str(), iPort, dWeight);

	// Add element to hash
	INT_32 iId = oMemcachedList.AddElement(szTMP, oMemcachedData, dWeight);

	// Get element
	MemcachedData & oTMP = oMemcachedList.GetElementById(iId);

	oTMP.hash_id = iId;

return iId;
}

//
// Get value from memcached
//
Memcached::eQueryResult ASMemcachedObject::Gets(const STLW::string  & sKey,
                                                CTPP::CDT           & oCDT,
                                                UINT_64             & iCASUnique)
{
	UINT_64       iFlags = 0;
	STLW::string  sValue;
	Memcached * pTMP = GetMemcachedServer(sKey);
	Memcached::eQueryResult oRC = pTMP -> Gets(sKey, sValue, iFlags, iCASUnique);
	if (oRC != Memcached::OK) { return oRC; }

	try
	{
		// Restore data
		if (sValue.size())
		{
			DecompressValue(sValue, iFlags);

			if (iFlags & 0x01) { RestoreCDT(sValue, oCDT); }
			else               { oCDT = sValue;            }
		}
	}
	catch(...)
	{
		pTMP -> Del(sKey);
		return Memcached::NOT_EXISTS;
	}

return oRC;
}

//
// Check And Set (CAS) value
//
Memcached::eQueryResult ASMemcachedObject::Cas(const STLW::string  & sKey,
                                               const CTPP::CDT     & oCDT,
                                               const UINT_64       & iCASUnique,
                                               INT_32                iExpires)
{
	// Serialize data
	STLW::string sSerializedValue;
	const UINT_64 iFlags = SerializeCDT(oCDT, sSerializedValue);

return GetMemcachedServer(sKey) -> Cas(sKey, sSerializedValue, iCASUnique, iFlags, iExpires);
}

//
// Get value from memcached
//
Memcached::eQueryResult ASMemcachedObject::Get(const STLW::string  & sKey,
                                               CTPP::CDT           & oCDT)
{
	UINT_64       iFlags = 0;
	STLW::string  sValue;
	Memcached * pTMP = GetMemcachedServer(sKey);
	Memcached::eQueryResult oRC = pTMP -> Get(sKey, sValue, iFlags);
	if (oRC != Memcached::OK) { return oRC; }

	try
	{
		// Restore data
		if (sValue.size())
		{
			DecompressValue(sValue, iFlags);
			if (iFlags & 0x01) { RestoreCDT(sValue, oCDT); }
			else               { oCDT = sValue;            }
		}
	}
	catch(...)
	{
		pTMP -> Del(sKey);
		return Memcached::NOT_EXISTS;
	}

return oRC;
}

//
// Set value
//
Memcached::eQueryResult ASMemcachedObject::Set(const STLW::string  & sKey,
                                               const CTPP::CDT     & oCDT,
                                               INT_32                iExpires)
{
	// Serialize data
	STLW::string sSerializedValue;
	UINT_64 iFlags = SerializeCDT(oCDT, sSerializedValue);

return GetMemcachedServer(sKey) -> Set(sKey, sSerializedValue, iFlags, iExpires);
}

//
// Add value
//
Memcached::eQueryResult ASMemcachedObject::Add(const STLW::string  & sKey,
                                               const CTPP::CDT     & oCDT,
                                               INT_32                iExpires)
{
	// Serialize data
	STLW::string sSerializedValue;
	UINT_64 iFlags = SerializeCDT(oCDT, sSerializedValue);

// Store data
return GetMemcachedServer(sKey) -> Add(sKey, sSerializedValue, iFlags, iExpires);
}

//
// Increment/add value
//
Memcached::eQueryResult ASMemcachedObject::Incr(const STLW::string  & sKey,
                                                INT_64              & iValue)
{
	return GetMemcachedServer(sKey) -> Incr(sKey, iValue);
}

//
// Decrement/substract value
//
Memcached::eQueryResult ASMemcachedObject::Decr(const STLW::string  & sKey,
                                                INT_64              & iValue)
{
	return GetMemcachedServer(sKey) -> Decr(sKey, iValue);
}

//
// Replace value
//
Memcached::eQueryResult ASMemcachedObject::Replace(const STLW::string  & sKey,
                                                   const CTPP::CDT     & oCDT,
                                                   INT_32                iExpires)
{
	// Serialize data
	STLW::string sSerializedValue;
	UINT_64 iFlags = SerializeCDT(oCDT, sSerializedValue);

// Store data
return GetMemcachedServer(sKey) -> Replace(sKey, sSerializedValue, iFlags, iExpires);
}

//
// Append data to string stored in cache
//
Memcached::eQueryResult ASMemcachedObject::Append(const STLW::string  & sKey,
                                                  const CTPP::CDT     & oCDT,
                                                  INT_32                iExpires)
{
	// Serialize data
	STLW::string sSerializedValue;
	SerializeCDT(oCDT, sSerializedValue);

// Store data, always as string
return GetMemcachedServer(sKey) -> Append(sKey, sSerializedValue, 0, iExpires);
}


//
// Prepend value to string stored in cache
//
Memcached::eQueryResult ASMemcachedObject::Prepend(const STLW::string  & sKey,
                                                   const CTPP::CDT     & oCDT,
                                                   INT_32                iExpires)
{
	// Serialize data
	STLW::string sSerializedValue;
	SerializeCDT(oCDT, sSerializedValue);

// Store data, always as string
return GetMemcachedServer(sKey) -> Prepend(sKey, sSerializedValue, 0, iExpires);
}

//
// Delete value from cache
//
Memcached::eQueryResult ASMemcachedObject::Del(const STLW::string & sKey)
{
	return GetMemcachedServer(sKey) -> Del(sKey);
}

//
// Connect to memcached server
//
Memcached::eQueryResult ASMemcachedObject::Connect(const INT_32 iServerId)
{
	if (iServerId >= oMemcachedList.Size()) { throw LogicError("Invalid server Id"); }

return oMemcachedList.GetElementById(iServerId).memcached -> Connect();
}

//
// Disconnect from memcached server
//
Memcached::eQueryResult ASMemcachedObject::Disconnect(const INT_32 iServerId)
{
	if (iServerId >= oMemcachedList.Size()) { throw LogicError("Invalid server Id"); }

return oMemcachedList.GetElementById(iServerId).memcached -> Disconnect();
}

//
// Disconnect & Connect
//
Memcached::eQueryResult ASMemcachedObject::Reconnect(const INT_32 iServerId)
{
	if (iServerId >= oMemcachedList.Size()) { throw LogicError("Invalid server Id"); }

return oMemcachedList.GetElementById(iServerId).memcached -> Reconnect();
}

//
// Connection status: OK or CLIENT_ERROR
//
Memcached::eQueryResult ASMemcachedObject::IsConnected(const INT_32 iServerId)
{
	if (iServerId >= oMemcachedList.Size()) { throw LogicError("Invalid server Id"); }

return oMemcachedList.GetElementById(iServerId).memcached -> IsConnected();
}

//
// Get Unix error code
//
INT_32 ASMemcachedObject::GetError(const INT_32 iServerId)
{
	if (iServerId >= oMemcachedList.Size()) { throw LogicError("Invalid server Id"); }

return oMemcachedList.GetElementById(iServerId).memcached -> GetError();
}

//
// Get object type
//
CCHAR_P ASMemcachedObject::GetObjectType() const { return "ASObject"; }

//
// Get object name
//
CCHAR_P ASMemcachedObject::GetObjectName() const { return "MemcachedConnector"; }

//
// Get Memcached server
//
Memcached * ASMemcachedObject::GetMemcachedServer(const STLW::string & sKey)
{
	if (oMemcachedList.Size() == 0)
	{
		throw LogicError("No available connections in pool");
	}

	for(;;)
	{
		// Get element
		MemcachedData & oMemcachedData = oMemcachedList.GetElementByKey(sKey);

		// Check connection
		if (oMemcachedData.memcached -> IsConnected() == Memcached::OK)
		{
			// All done
			return oMemcachedData.memcached;
		}
		// Not connected?
		else /* if (oMemcachedData.memcached -> IsConnected() == Memcached::NOT_CONNECTED) */
		{
			// Get current time
			struct timeval oCurrentTS;
			gettimeofday(&oCurrentTS, NULL);

			UINT_64 iCurrentTS = oCurrentTS.tv_sec * 10000 + oCurrentTS.tv_usec / 100;

			// Check disconnection time and wait timeout
			if (oMemcachedData.last_error_ts + iErrorTimeout < iCurrentTS)
			{
				// Ok memcached server is now reconnected
				Memcached::eQueryResult oRC = oMemcachedData.memcached -> Connect();
				if (oRC == Memcached::OK)
				{
//fprintf(stderr, "Enable element %d\n", oMemcachedData.hash_id);
					// Enable element
					oMemcachedList.EnableElementById(oMemcachedData.hash_id);
					// All done
					return oMemcachedData.memcached;
				}
//fprintf(stderr, "Disable element %d\n", oMemcachedData.hash_id);
				// Not OK?
				oMemcachedData.last_error_ts = iCurrentTS;
				// Disable element
				oMemcachedList.DisableElementById(oMemcachedData.hash_id);
			}
		}
	}
// This should *never* happened
return NULL;
}

//
// Get Memcache object by id
//
Memcached * ASMemcachedObject::GetMemcachedById(const UINT_32 iPos)
{
	if (iPos >= vServers.size()) { return NULL; }

return vServers[iPos];
}

//
// Get number or servers
//
UINT_32  ASMemcachedObject::GetServersNumber() const { return vServers.size(); }

//
// A destructor
//
ASMemcachedObject::~ASMemcachedObject() throw()
{
	STLW::vector<Memcached *>::iterator itvServers = vServers.begin();
	while (itvServers != vServers.end())
	{
		delete *itvServers;
		++itvServers;
	}
}

} // namespace CAS_MOD
// End.
